package com.atguigu.hbase;

import javafx.scene.paint.Stop;
import lombok.Getter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class HBaseUtil {
    @Getter
    private  static Connection connection;
    static {
        Configuration configuration = new Configuration();
        configuration.set("hbase.zookeeper.quorum" , "hadoop102,hadoop103,hadoop104");
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void closeConnection(Connection connection) throws IOException {
        connection.close();
    }

    // TODO DDL 建表
    public static void createTable(Connection connection, String namespaceName, String tableName, String ...cfs) throws IOException {
        if(connection == null) {
            System.out.println("连接对象不能为空");
            return;
        }
        if(tableName == null || tableName.trim().isEmpty()) {
            System.out.println("表名不能为空");
            return;
        }
        if(cfs == null || cfs.length == 0) {
            System.out.println("至少指定一个列族");
            return;
        }

        Admin admin = connection.getAdmin();
        TableName tn = TableName.valueOf(namespaceName, tableName);

        // 判断表是否已存在
        boolean tableExists = admin.tableExists(tn);
        if(tableExists) {
            System.out.println("表已存在");
            return;
        }

        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tn);
        //设置列族
        for (String cf : cfs) {
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
            ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor) ;
        }
        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();

        admin.createTable(tableDescriptor);

        System.out.println( (namespaceName == null ? "default":namespaceName) + ":" + tableName + "创建成功");

        admin.close();
    }

    // TODO 删表
    public static void deleteTable(Connection connection, String namespaceName, String tableName) throws IOException {
        if(connection == null) {
            System.out.println("连接对象为空");
            return;
        }
        if(tableName == null || tableName.trim().isEmpty()) {
            System.out.println("表名不能为空");
        }

        Admin admin = connection.getAdmin();
        TableName tn = TableName.valueOf(namespaceName, tableName);
        boolean tableExists = admin.tableExists(tn);
        if(!tableExists) {
            System.out.println("表不存在");
            return;
        } else {
            admin.disableTable(tn);
            admin.deleteTable(tn);
            System.out.println("表已删除");
        }
    }

    // TODO list
    public static void listTable(Connection connection) throws IOException {
        if(connection == null) {
            System.out.println("连接对象为空");
            return;
        }
        Admin admin = connection.getAdmin();
        List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();
        for (TableDescriptor tableDescriptor : tableDescriptors) {
            System.out.println(tableDescriptor);
        }
    }

    // TODO 加列族
    public static void addColumnFamily(Connection connection,String namespaceName, String tableName, String ...cfs) throws IOException {
        if(connection == null) {
            System.out.println("连接对象为空");
            return;
        }
        Admin admin = connection.getAdmin();
        TableName tn = TableName.valueOf(namespaceName, tableName);
        for (String cf : cfs) {
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder =
                    ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
            ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
            admin.addColumnFamily(tn, columnFamilyDescriptor);
        }

        admin.close();
    }

    // TODO DML: put 'stu', '1001', 'cf:f1', 20
    public static void putData(Connection connection,String namespaceName, String tableName,
                               String rowKey, String cf, String col, String value) throws IOException {
        // 判空
        checkEmpty(connection);
        checkEmpty(tableName,rowKey,cf,col,value);

        // Table
        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);

        // put
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(col),Bytes.toBytes(value));
        table.put(put);

        table.close();
    }

    // TODO DML 删除
    public static void deleteData(Connection connection,String namespaceName, String tableName,
                               String rowKey, String cf, String col) throws IOException {
        checkEmpty(connection);
        checkEmpty(tableName,rowKey,cf,col);

        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);

        Delete delete = new Delete(Bytes.toBytes(rowKey));
        // 删除 delete效果 删除一条 暴漏出旧数据
//        delete.addColumn(Bytes.toBytes(cf), Bytes.toBytes(col));

        // 删除指定列 deleteColumn
//        delete.addColumns(Bytes.toBytes(cf),Bytes.toBytes(col));

        // 删除指定列族 deleteFamily
//        delete.addFamily(Bytes.toBytes(cf));

        // delete什么都不加 删除一行数据 所有列族
        table.delete(delete);

        table.close();

    }

    // TODO DML get
    public static void getData(Connection connection, String namespaceName, String tableName,String rowKey) throws IOException {
        checkEmpty(connection);
        checkEmpty(tableName,rowKey);

        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);

        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);

        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            // 处理cell 从cell中提取信息
            String kv = Bytes.toString(CellUtil.cloneRow(cell)) + ":" +
                    Bytes.toString(CellUtil.cloneFamily(cell)) + ":" +
                    Bytes.toString(CellUtil.cloneQualifier(cell)) + ":" +
                    Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println(kv);
        }

        table.close();
    }

    // TODO DML scan
    public static void scanData(Connection connection, String namespaceName, String tableName, String startRow,String stopRow) throws IOException{
        checkEmpty(connection);
        checkEmpty(tableName,startRow,stopRow);

        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);

        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(startRow)).withStopRow(Bytes.toBytes(stopRow));
        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String kv = Bytes.toString(CellUtil.cloneRow(cell)) + ":" +
                        Bytes.toString(CellUtil.cloneFamily(cell)) + ":" +
                        Bytes.toString(CellUtil.cloneQualifier(cell)) + ":" +
                        Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(kv);
            }
            System.out.println("---------------");
        }
        table.close();
    }

    // TODO scan with filter 类似where过滤 基于某个列 效率低 要做全表扫描
    public static void scanDataWithFilter(Connection connection, String namespaceName, String tableName) throws IOException {
        checkEmpty(connection);
        checkEmpty(tableName);
        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);

        Scan scan = new Scan();

        SingleColumnValueFilter nameFilter = new SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"),
                CompareOperator.EQUAL, Bytes.toBytes("tom"));

        SingleColumnValueFilter ageFilter = new SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("age"),
                CompareOperator.GREATER_OR_EQUAL, Bytes.toBytes("15"));

        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, nameFilter, ageFilter);
        scan.setFilter(filterList);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //处理Cell ， 从Cell中提取列相关的信息
                String kv = Bytes.toString(CellUtil.cloneRow(cell)) + " : " +
                        Bytes.toString(CellUtil.cloneFamily(cell)) + " : " +
                        Bytes.toString(CellUtil.cloneQualifier(cell)) + " : " +
                        Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(kv);
            }

            System.out.println("---------------------------------");
        }


        table.close();
    }


    public static void checkEmpty(Connection connection) {
        if(connection == null) {
            System.out.println("连接不能为空");
            return;
        }
    }
    public static void checkEmpty(String ...params) {
        for (String param : params) {
            if(param == null || param.trim().isEmpty()) {
                System.out.println("参数" + param + "不能为空");
                return;
            }
        }
    }


    public static void main(String[] args) throws IOException {
        Connection connection = getConnection();
//        createTable(connection, null, "test", "f1", "f2");
//        deleteTable(connection, "mydb", "t1");
//        listTable(connection);
//        addColumnFamily(connection, null, "stu","f3","f4");

//        putData(connection,null,"stu","1002","f1","name","jack");
//        putData(connection,null,"stu","1002","f1","age","23");
//        putData(connection,null,"stu","1002","f2","addr","beijing");
//        putData(connection,null,"stu","1002","f2","gender","f");
//
//        putData(connection,null,"stu","1003","f1","name","david");
//        putData(connection,null,"stu","1003","f1","age","23");
//        putData(connection,null,"stu","1003","f2","addr","nanjing");
//        putData(connection,null,"stu","1003","f2","gender","f");

//        putData(connection,null,"stu","1004","f1","name","tom");
//        putData(connection,null,"stu","1004","f1","age","18");
//        putData(connection,null,"stu","1004","f2","addr","nanjing");
//        putData(connection,null,"stu","1004","f2","gender","f");

//        deleteData(connection,null,"stu","1001","f4","age");
//        getData(connection,null,"stu","1001");
//        scanData(connection,null,"stu","1001", "1003");
        scanDataWithFilter(connection,null,"stu");
        closeConnection(connection);
    }

}
